package com.zhh.nio;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class NioServer {

    protected ServerSocketChannel serverSock;
    /**
     * Address for the server socket.
     */
    private InetAddress address;
    
    private SessionHandler sessionHandler;
    private BlockingQueue<NioSession> register;
    
    public enum  EVENT_TYPE {
        READ ,WRITE
    }

    public InetAddress getAddress() {
        return address;
    }

    public void setAddress(InetAddress address) {
        this.address = address;
    }

    /**
     * Allows the server developer to specify the backlog that should be used
     * for server sockets. By default, this value is 100.
     */
    private int backlog = 100;

    public void setBacklog(int backlog) {
        if (backlog > 0)
            this.backlog = backlog;
    }

    public int getBacklog() {
        return backlog;
    }

    /**
     * Server socket port.
     */
    private int port;

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    // poller的默认为cpu的核数
    private int pollerThreadCount = Runtime.getRuntime().availableProcessors();

    // selector.select()的time out
    private long selectTimeout = 1000L;

    private AtomicInteger pollerRotater = new AtomicInteger(0);

    private Poller[] pollers;
    private Acceptor acceptor;

    /**
     * Running state of the endpoint.
     */
    protected volatile boolean running = false;

    /**
     * Will be set to true whenever the endpoint is paused.
     */
    protected volatile boolean paused = false;

    protected Poller getPoller0() {

        // Thread.currentThread().getId()
        int idx = Math.abs(pollerRotater.incrementAndGet()) % pollers.length;
        return pollers[idx];
    }
    
    public SessionHandler getSessionHandler() {
        return sessionHandler;
    }

    public void setSessionHandler(SessionHandler sessionHandler) {
        this.sessionHandler = sessionHandler;
    }

    protected class Poller implements Runnable {

        private Selector selector;

        protected volatile int keyCount = 0;

        public Poller() throws Exception {

            synchronized (Selector.class) {
                selector = Selector.open();
            }
        }

        public Selector getSelector() {
            return selector;
        }

        public void setSelector(Selector selector) {
            this.selector = selector;
        }

        public void register(NioSession session) throws IOException {

            session.setPoller(this);
            register.offer(session);
            selector.wakeup();
        }
        
        public void handlerRegister() throws IOException{
            NioSession item = null;
            while((item = register.poll()) != null){
                item.register();
            }
        }

        public void run() {

            while (true) {

                try {
                    handlerRegister();
                    keyCount = selector.select(selectTimeout);
                    if(keyCount > 0)
                        debug(Thread.currentThread().getName() + " selected:" + keyCount);
                    Iterator<SelectionKey> iterator =
                            keyCount > 0 ? selector.selectedKeys().iterator() : null;
                        // Walk through the collection of ready keys and dispatch
                        // any active event.
                        while (iterator != null && iterator.hasNext()) {
                            SelectionKey sk = iterator.next();
                            NioSession session = (NioSession)sk.attachment();
                            // Attachment may be null if another thread has called
                            // cancelledKey()
                            if (session == null) {
                                iterator.remove();
                            } else {
                                iterator.remove();
                                if(sk.isValid() && sk.isWritable()){
                                    debug("Session write:" + session);
                                    session.write();
                                }else if(sk.isValid() && sk.isReadable()){
                                    debug("Session read:" + session);
                                    session.read();
                                }else{
                                	debug("Session close from select");
                                    session.close();
                                }
                            }
                        }//while
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

        }

    }
    

    public void bind() throws Exception {

        serverSock = ServerSocketChannel.open();
        // socketProperties.setProperties(serverSock.socket());
        InetSocketAddress addr = (getAddress() != null ? new InetSocketAddress(
                getAddress(), getPort()) : new InetSocketAddress(getPort()));
        serverSock.socket().bind(addr, getBacklog());
        serverSock.configureBlocking(true); // mimic APR behavior
    }
    
    public NioServer(int port){
        
        this.port = port;
        try {
            bind();
            acceptor = new Acceptor();
            register = new ArrayBlockingQueue<NioSession>(10);
           // new Thread(acceptor).start();
            //pollerThreadCount = 1;//debug使用
            pollers = new Poller[pollerThreadCount];
           for(int i = 0; i < pollerThreadCount; i++){
               pollers[i] = new Poller();
              // new Thread(pollers[i]).start();
           }
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        
    }
    
    public void start(){
        
        running = true;
        new Thread(acceptor,"Acceptor").start();
        for(int i = 0; i < pollerThreadCount; i++){
           new Thread(pollers[i],"Poller-" + i ).start();
        }
    }
    
    
    public static void main(String[] args){
        
       NioServer server = new NioServer(8888);
       SessionHandler sessionHandler = new DefaultSessionHandler();
       server.setSessionHandler(sessionHandler);
       server.start();
    }

    // 第一版搭框架，不会使用pool等功能
    public void handleNewSession(SocketChannel socket) throws IOException {
         //set blocking  to false
        socket.configureBlocking(false);
        socket.socket().setSoTimeout(14);
        NioSession session = new NioSession(socket);
        session.setSessionHandler(sessionHandler);
        getPoller0().register(session);
    }

    protected class Acceptor implements Runnable {

        public void run() {

            while (running) {

                SocketChannel socket = null;
                try {
                    socket = serverSock.accept();
                    debug("receive a socket:" + socket);
                    if (running) {
                         handleNewSession(socket);
                    }
                } catch (IOException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

            }

        }

    }
    
    private void debug(String message){
        
        System.out.println(message);
    }

}
